package com.gitee.wsl.flowing.collector

import com.gitee.wsl.ioDispatcher
import kotlinx.coroutines.CoroutineDispatcher
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Dispatchers
import kotlinx.coroutines.channels.Channel
import kotlinx.coroutines.channels.ReceiveChannel
import kotlinx.coroutines.channels.SendChannel
import kotlinx.coroutines.launch
import kotlinx.coroutines.selects.select


/**
 * The engine of the MapView. The view-model uses two channels to communicate with the [TileCollector]:
 * * one to send [TileSpec]s (a [SendChannel])
 * * one to receive [Tile]s (a [ReceiveChannel])
 *
 * The [TileCollector] encapsulates all the complexity that transforms a [TileSpec] into a [Tile].
 * ```
 *                                              _____________________________________________________________________
 *                                             |                           TileCollector             ____________    |
 *                                  tiles      |                                                    |  ________  |   |
 *              ---------------- [*********] <----------------------------------------------------- | | worker | |   |
 *             |                               |                                                    |  --------  |   |
 *             ↓                               |                                                    |  ________  |   |
 *  _____________________                      |                                  tileSpecs         | | worker | |   |
 * | TileCanvasViewModel |                     |    _____________________  <---- [**********] <---- |  --------  |   |
 *  ---------------------  ----> [*********] ----> | tileCollectorKernel |                          |  ________  |   |
 *                                tileSpecs    |    ---------------------  ----> [**********] ----> | | worker | |   |
 *                                             |                                  tileSpecs         |  --------  |   |
 *                                             |                                                    |____________|   |
 *                                             |                                                      worker pool    |
 *                                             |                                                                     |
 *                                              ---------------------------------------------------------------------
 * ```
 * This architecture is an example of Communicating Sequential Processes (CSP).
 *
 * @author p-lr on 22/06/19
 */
class Collector<TileSpec,Tile,Bitmap>(private val workerCount: Int,
                                      val dispatcher:CoroutineDispatcher = Dispatchers.ioDispatcher ,
                                      val download:suspend (TileSpec) -> Bitmap,
                                      val processing:suspend (Bitmap) -> Tile?
){
    /**
     * Sets up the tile collector machinery. The architecture is inspired from
     * [Kotlin Conf 2018](https://www.youtube.com/watch?v=a3agLJQ6vt8).
     * It support back-pressure, and avoids deadlock in CSP taking into account recommendations of
     * this [article](https://medium.com/@elizarov/deadlocks-in-non-hierarchical-csp-e5910d137cc),
     * which is from the same author.
     *
     * @param [tileSpecs] channel of [TileSpec], which capacity should be [Channel.RENDEZVOUS].
     * @param [tilesOutput] channel of [Tile], which should be set as [Channel.RENDEZVOUS].
     */
    fun CoroutineScope.collectTiles(tileSpecs: ReceiveChannel<TileSpec>,
                                    tilesOutput: SendChannel<Tile>
    ) {
        val tilesToDownload = Channel<TileSpec>(capacity = Channel.RENDEZVOUS)
        val tilesDownloadedFromWorker = Channel<TileSpec>(capacity = 1)

        repeat(workerCount) { worker(tilesToDownload, tilesDownloadedFromWorker, tilesOutput) }
        tileCollectorKernel(tileSpecs, tilesToDownload, tilesDownloadedFromWorker)
    }

    private fun CoroutineScope.worker(tilesToDownload: ReceiveChannel<TileSpec>,
                                      tilesDownloaded: SendChannel<TileSpec>,
                                      tilesOutput: SendChannel<Tile>
                                      ) = launch(dispatcher) {
        for (spec in tilesToDownload) {
            val bitmap = download(spec)
            tilesDownloaded.send(spec)
            val tile = processing(bitmap) ?: continue
            tilesOutput.send(tile)
        }
    }

    private fun CoroutineScope.tileCollectorKernel(tileSpecs: ReceiveChannel<TileSpec>,
                                                   tilesToDownload: SendChannel<TileSpec>,
                                                   tilesDownloadedFromWorker: ReceiveChannel<TileSpec>) = launch(Dispatchers.Default) {

        val tilesBeingProcessed = mutableListOf<TileSpec>()

        while (true) {
            select<Unit> {
                tilesDownloadedFromWorker.onReceive {
                    tilesBeingProcessed.remove(it)
                }
                tileSpecs.onReceive {
                    if (!tilesBeingProcessed.any { spec -> spec == it }) {
                        /* Add it to the list of locations being processed */
                        tilesBeingProcessed.add(it)

                        /* Now download the tile */
                        tilesToDownload.send(it)
                    }
                }
            }
        }
    }

}